package event

import (
	"encoding/hex"
	"gitee.com/go-vulcanus/vulcanus"
	"gitee.com/go-vulcanus/vulcanus/conf"
	"gitee.com/go-vulcanus/vulcanus/data"
	"gitee.com/go-vulcanus/vulcanus/session"
	"gitee.com/go-vulcanus/vulcanus/utils"
	"gitee.com/go-vulcanus/vulcanus/utils/date"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/google/wire"
	"github.com/panjf2000/ants/v2"
	"sync"
	"time"
)

var EventProviderSet = wire.NewSet(NewEventListener)

var event *Event

type Event struct {
	da           *data.Data
	sm           *session.SessionManager
	channel      chan interface{}
	wchannel     chan interface{}
	replyChannel sync.Map
	traceMsg     bool
	p            *ants.PoolWithFunc
	waiting      sync.Map
}

type taskArgs struct {
	deviceCode string
	times      int
	waitingSec int64
	f          func()
}

func NewEventListener(app *conf.App, da *data.Data, sm *session.SessionManager) *Event {
	e := &Event{
		channel:  make(chan interface{}, 1000),
		wchannel: make(chan interface{}, 1000),
		da:       da,
		sm:       sm,
		traceMsg: app.Device.TraceMsg,
	}

	p, _ := ants.NewPoolWithFunc(1000, e.resend)
	e.p = p

	event = e

	return e
}

func (e *Event) Listen() {
	go func() {
		for {
			e.listenEvent()
		}
	}()

	go func() {
		for ev := range e.wchannel {
			replyEvent := ev.(vulcanus.MqttReplyEvent)

			if replyEvent.NeedReply {
				if _, ok := e.waiting.Load(replyEvent.DeviceConfig.DeviceCode); ok {
					// 当前正在等待设备回复
					time.AfterFunc(time.Second*5, func() {
						e.wchannel <- ev
					})
					continue
				}

				args := &taskArgs{
					deviceCode: replyEvent.DeviceConfig.DeviceCode,
					waitingSec: 5,
					times:      3,
					f: func() {
						e.consumeMqttReplyEvent(replyEvent)
					},
				}
				e.waiting.Store(replyEvent.DeviceConfig.DeviceCode, struct{}{})
				e.p.Invoke(args)
			} else {
				time.AfterFunc(time.Second*1, func() {
					e.consumeMqttReplyEvent(replyEvent)
				})
			}
		}
	}()
}

func (e *Event) resend(a interface{}) {
	args := a.(*taskArgs)

	if args.times == 0 {
		e.waiting.Delete(args.deviceCode)
		return
	}

	if _, ok := e.replyChannel.Load(args.deviceCode); !ok {
		e.replyChannel.Store(args.deviceCode, make(chan string, 1000))
	}

	args.f()

	c, _ := e.replyChannel.Load(args.deviceCode)

	select {
	case _ = <-c.(chan string):
		e.waiting.Delete(args.deviceCode)
		return
	case <-time.After(time.Duration(args.waitingSec) * time.Second):
		// time out, resend msg
		args.times -= 1
		args.waitingSec = args.waitingSec * 4
		e.resend(args)
	}
}

func Reply(deviceCode string, synInf string) {
	if c, ok := event.replyChannel.Load(deviceCode); ok {
		c.(chan string) <- synInf
	}
}

// listenEvent start event listening.
func (e *Event) listenEvent() {
	var deviceCode string
	var errEvent interface{}

	defer func() {
		if err := recover(); err != nil {
			log.Errorf("[%s] event [%v] consume err: [%s]", deviceCode, errEvent, err)
		}
	}()

	for ev := range e.channel {
		errEvent = ev
		switch ev.(type) {
		case vulcanus.SensorDataEvent:
			deviceCode = ev.(vulcanus.SensorDataEvent).DeviceConfig.DeviceCode
			e.consumeSensorDataEvent(ev.(vulcanus.SensorDataEvent))
			break
		case vulcanus.ControlRespEvent:
			break
		//case vulcanus.MqttReplyEvent:
		//	deviceCode = ev.(vulcanus.MqttReplyEvent).DeviceConfig.DeviceCode
		//	e.consumeMqttReplyEvent(ev.(vulcanus.MqttReplyEvent))
		//	break
		case vulcanus.StateDataEvent:
			deviceCode = ev.(vulcanus.StateDataEvent).DeviceConfig.DeviceCode
			e.consumeStateDataEvent(ev.(vulcanus.StateDataEvent))
			break
		case vulcanus.ControllerStateEvent:
			deviceCode = ev.(vulcanus.ControllerStateEvent).DeviceConfig.DeviceCode
			e.consumeControllerStateEvent(ev.(vulcanus.ControllerStateEvent))
			break
		case vulcanus.SocketSendEvent:
			deviceCode = ev.(vulcanus.SocketSendEvent).DeviceConfig.DeviceCode
			e.consumeSocketSendEvent(ev.(vulcanus.SocketSendEvent))
			break
		}
	}
}

func PublishSensorDataEvent(sensorData vulcanus.SensorDataEvent) {
	event.channel <- sensorData
}

func PublishControlRespEvent(controlResp vulcanus.ControlRespEvent) {
	event.channel <- controlResp
}

func PublishMqttReplyEvent(mqttReply vulcanus.MqttReplyEvent) {
	event.wchannel <- mqttReply
}

func PublishStateDataEvent(stateData vulcanus.StateDataEvent) {
	event.channel <- stateData
}

func PublishControllerStateEvent(controllerState vulcanus.ControllerStateEvent) {
	event.channel <- controllerState
}

func PublishSocketSendEvent(sockSend vulcanus.SocketSendEvent) {
	event.channel <- sockSend
}

func (e *Event) consumeSensorDataEvent(sensorData vulcanus.SensorDataEvent) {
	if e.traceMsg {
		log.Infof("[%s] sensor data: %v", sensorData.DeviceConfig.DeviceCode, sensorData.Data)
	}
	sensorConfigList := sensorData.DeviceConfig.SensorConfigList
	if len(sensorConfigList) > len(sensorData.Data) {
		log.Warnf("[%s] sensor config size more than data size.", sensorData.DeviceConfig.DeviceCode)
		return
	}
	data := make([]string, len(sensorConfigList))

	if sensorData.DeviceConfig.DeviceType == "WD-GHC300" {
		for i := range sensorConfigList {
			data[i] = sensorData.Data[i]
		}
	} else {
		for i, sensorConfig := range sensorConfigList {
			data[i] = sensorData.Data[sensorConfig.Num]
		}
	}

	sd := &vulcanus.SensorData{
		Time:       sensorData.Time,
		SystemTime: date.Now(),
		DeviceTime: sensorData.Time,
		Data:       data,
	}
	deviceData := vulcanus.NewDeviceData(sensorData.DeviceConfig, sd)
	e.da.Kp.SendDeviceData(utils.ToJsonString(deviceData))
}

func (e *Event) consumeStateDataEvent(stateData vulcanus.StateDataEvent) {
	if e.traceMsg {
		log.Infof("[%s] state data: %v", stateData.DeviceConfig.DeviceCode, stateData.Data)
	}
	sd := &vulcanus.StateData{
		Time: stateData.Time,
		Data: stateData.Data,
	}
	deviceStateData := vulcanus.NewDeviceStateData(stateData.DeviceConfig, sd)
	e.da.Kp.SendStateData(utils.ToJsonString(deviceStateData))
}

func (e *Event) consumeMqttReplyEvent(mqttReply vulcanus.MqttReplyEvent) {
	if e.traceMsg {
		log.Infof("[%s] mqtt reply topic: [%s], content: [%s]", mqttReply.DeviceConfig.DeviceCode, mqttReply.Topic, mqttReply.EncodedContent)
	}
	channel := mqttReply.DeviceChannel
	channel.WriteToMqtt(mqttReply.Topic, mqttReply.EncodedContent)
	// 向下游发送设备原始数据
	metadata := &vulcanus.Metadata{
		DeviceCode: mqttReply.DeviceConfig.DeviceCode,
		Timestamp:  date.NowMs(),
		Type:       "发送数据",
		Data:       mqttReply.Content,
	}
	e.da.Kp.SendMetadata(utils.ToJsonString(metadata))
}

func (e *Event) consumeControllerStateEvent(controllerState vulcanus.ControllerStateEvent) {
	if e.traceMsg {
		log.Infof("[%s] controller state: %v", controllerState.DeviceConfig.DeviceCode, controllerState.Openings)
	}
	cs := vulcanus.ControllerState{
		DeviceCode:   controllerState.DeviceConfig.DeviceCode,
		DeviceType:   controllerState.DeviceConfig.DeviceType,
		SwitchStatus: controllerState.SwitchStatus,
		Openings:     controllerState.Openings,
		Modes:        controllerState.Modes,
	}
	e.da.Kp.SendControllerState(utils.ToJsonString(cs))
}

func (e *Event) consumeSocketSendEvent(event vulcanus.SocketSendEvent) {
	hexContent := hex.EncodeToString(event.Content)
	if e.traceMsg {
		log.Infof("[%s] socket send: %v", event.DeviceConfig.DeviceCode, hexContent)
	}
	deviceChannel := e.sm.Get(event.DeviceConfig.DeviceCode)
	if deviceChannel != nil {
		deviceChannel.WriteToSocket(event.Content)
	}
	// 向下游发送设备原始数据
	metadata := &vulcanus.Metadata{
		DeviceCode: event.DeviceConfig.DeviceCode,
		Timestamp:  date.NowMs(),
		Type:       "发送数据",
		Data:       hexContent,
	}
	e.da.Kp.SendMetadata(utils.ToJsonString(metadata))
}
